package rabbitmq.component;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;
import com.rabbitmq.client.Channel;

import java.io.IOException;


/**
 * 接收mq消息
 */
@Component
public class RabbitReceiverHello {
    public Logger log = LoggerFactory.getLogger(this.getClass());

    @RabbitListener(queues = "hello")
    public void helloQueue(Message message, Channel channel,@Header(AmqpHeaders.DELIVERY_TAG) long tag) {
       log.info(" helloQueue接收消息 :{} {} " ,tag, new String(message.getBody()));
        try {
            /**
             * void basicQos(int prefetchSize, int prefetchCount, boolean global) throws IOException;
             * prefetchSize：0，单条消息大小限制，0代表不限制
             * prefetchCount：一次性消费的消息数量。会告诉 RabbitMQ 不要同时给一个消费者推送多于 N 个消息，
             *  即一旦有 N 个消息还没有 ack，则该 consumer 将 block 掉，直到有消息 ack。
             * global：true、false 是否将上面设置应用于 channel，简单点说，就是上面限制是 channel 级别的还是 consumer 级别。
             *  当我们设置为 false 的时候生效，设置为 true 的时候没有了限流功能，因为 channel 级别尚未实现。
             */
            channel.basicQos(0, 3, false);
            /**
             * deliveryTag:该消息的index
             *  multiple：是否批量.true:将一次性ack所有小于deliveryTag的消息。
             */
            channel.basicAck(tag, false);
           // int a =  1/0;
        }catch (Exception e){
            log.error("接收-error");
            try {
                /**
                 * deliveryTag:该消息的index
                 * multiple：是否批量.true:将一次性拒绝所有小于deliveryTag的消息。
                 * requeue：被拒绝的是否重新入队列
                 */
                channel.basicNack(tag,false,true);
                /**
                 * deliveryTag:该消息的index
                 *  requeue：被拒绝的是否重新入队列
                 */
               // channel.basicReject(tag:, false);
            } catch (Exception ex) {
                log.error(e.getMessage());
            }
        }
    }



    @RabbitListener(bindings=@QueueBinding(value=@Queue(value="fanout2",durable = "true",autoDelete="false"),
            exchange=@Exchange(value="fanout.exchange",durable = "true",type= ExchangeTypes.FANOUT)))
    public void fanoutExchange(Message message,Channel channel,@Header(AmqpHeaders.DELIVERY_TAG) long tag) {
        log.info("fanout接收消息  : " + new String(message.getBody()));
        try {
            channel.basicAck(tag, false);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }


    @RabbitListener(bindings=@QueueBinding(value=@Queue(value="direct2",durable = "true",autoDelete="false"),
            exchange=@Exchange(value="direct.exchange",durable = "true",type=ExchangeTypes.DIRECT), key="routingKey.direct"))
    public void directExchange(Message message,Channel channel,@Header(AmqpHeaders.DELIVERY_TAG) long tag) {
        log.info("direct接收消息  : " + new String(message.getBody()));
        try {
            channel.basicAck(tag, false);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }


    @RabbitListener(queues = "timeout.queue")
    public void timeout(Message message, Channel channel,@Header(AmqpHeaders.DELIVERY_TAG) long tag) {
        log.info("timeout接收消息 :{} {} " ,tag, new String(message.getBody()));
        try {
            channel.basicAck(tag, false);
        }catch (Exception e){
            log.error("接收-error");
        }
    }

}


